package com.chief.state.exactlyonce;

import com.chief.state.Data;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Properties;

public class FlinkKafkaExactlyOnce {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env.setStateBackend(new HashMapStateBackend());

        env.getCheckpointConfig().setCheckpointInterval(10000);
        env.getCheckpointConfig().setCheckpointStorage("viewfs://mycluster/data1/flink-cp/");
        env.getCheckpointConfig().setCheckpointTimeout(50000);

        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, org.apache.flink.api.common.time.Time.seconds(20)));

        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers("hadoop53:9092,hadoop54:9092,hadoop55:9092")
                .setTopics("inTest")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();


        //张三,2022-03-29 10:00:00,50.1
        SingleOutputStreamOperator<Data> map = env.fromSource(kafkaSource,
                WatermarkStrategy.noWatermarks(),
                "kafka")
                .map((MapFunction<String, Data>) value -> {
                    String[] split = value.split(",");
                    //当大于100，休眠20s
                    if (Double.valueOf(split[2]).compareTo(Double.valueOf(100)) > 0) {
                        Thread.sleep(20000);
                    }
                    return new Data(split[0], DateUtils.parseDate(split[1], "yyyy-MM-dd HH:mm:ss"), Double.valueOf(split[2]));
                });

        SingleOutputStreamOperator<Data> watermarks = map.assignTimestampsAndWatermarks(WatermarkStrategy.<Data>forBoundedOutOfOrderness(Duration.ofMillis(200))
                .withTimestampAssigner((element, recordTimestamp) -> element.getDate().getTime()));

        SingleOutputStreamOperator<Data> sum = watermarks.keyBy(Data::getName)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .sum("score");

        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop53:9092,hadoop54:9092,hadoop55:9092");
        properties.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,"100000");

//        sum.print();

        KafkaSerializationSchema<Data> serializationSchema = (element, timestamp) -> {
            return new ProducerRecord<>(
                    "outTest", // target topic
                    element.toString().getBytes(StandardCharsets.UTF_8)); // record contents
        };

        FlinkKafkaProducer<Data> myProducer = new FlinkKafkaProducer<>(
                "outTest",
                serializationSchema,
                properties,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE);

        sum.addSink(myProducer);

        env.execute("FlinkKafkaExactlyOnce");
    }
}
